package com.pw.gmall.realtime.utils;

import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.TableProcess;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @Author: linux_future
 * @since: 2022/3/15
 **/
public class KafkaUtils {
public static FlinkKafkaConsumer<String> getKafkaSource(String groupName,String topicName){

    Properties props = new Properties();
    props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
    props.put("group.id", groupName);
    props.put("auto.offset.reset","latest");
    props.put("isolation.level","read_committed");


    return new FlinkKafkaConsumer<String>(
            topicName,
            new SimpleStringSchema(),
            props
    );
}

    public static SinkFunction<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);


        return new FlinkKafkaProducer<String>(
                "default",
                new KafkaSerializationSchema<String>() {

                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                    @Nullable Long timestamp) {
                        return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.put("transaction.timeout.ms", 15 * 60 * 1000);

        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
                "default",
                new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> element,
                                                                    @Nullable Long timestamp) {
                        String topic = element.f1.getSink_table();
                        byte[] data = element.f0.toJSONString().getBytes(StandardCharsets.UTF_8);
                        return new ProducerRecord<>(topic, data);
                    }
                },
                props,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }

}
